package com.demo.kettle.test.service.impl;

import com.demo.kettle.test.service.IKettleTestService;
import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.cluster.SlaveServer;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.Result;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleMissingPluginsException;
import org.pentaho.di.core.exception.KettleXMLException;
import org.pentaho.di.core.variables.Variables;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobExecutionConfiguration;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransExecutionConfiguration;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.www.SlaveServerJobStatus;
import org.pentaho.di.www.SlaveServerTransStatus;
import org.springframework.core.io.FileSystemResource;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 服务实现类
 *
 * @author yuanzheng
 * @since 2022-02-28
 */
@Service
@Slf4j
public class KettleTestServiceImpl implements IKettleTestService {

    @Override
    public String job() throws KettleException, ExecutionException, InterruptedException {

        // 初始化执行环境
        KettleEnvironment.init();

        AtomicBoolean jobStarted = new AtomicBoolean(true);
        int countOfCheckers = 100;
        HashMap<Integer, Future<StringBuilder>> checkers = new HashMap<>(16);

        String testJobFullPath = "E:\\yuanzheng-codebase\\m-framework\\kettle-git\\file\\test.kjb";
        JobMeta jm = new JobMeta(testJobFullPath, null);
        final Job job = new Job(null, jm);

        ExecutorService executorService = Executors.newFixedThreadPool(countOfCheckers);
        for (int i = 0; i < countOfCheckers; i++) {
            Future<StringBuilder> future = executorService.submit(new Callable<StringBuilder>() {
                private StringBuilder statusCollector = new StringBuilder();

                @Override
                public StringBuilder call() throws Exception {
                    while (jobStarted.get()) {
                        statusCollector.append(job.getStatus());
                        //add the sleep to reduce the count of logged status and prevent OOM
                        Thread.sleep(10);
                    }
                    return statusCollector;
                }
            });
            checkers.put(i, future);
        }
        job.start();
        job.waitUntilFinished();
        jobStarted.set(false);

        for (Future<StringBuilder> checker : checkers.values()) {
            System.out.println(checkOrderStatus(checker.get().toString()));
        }

        // 关闭执行器
        executorService.shutdown();

        // 关闭执行环境
        KettleEnvironment.shutdown();
        return null;
    }

    @Override
    public String trans() throws KettleException {

        // 初始化运行环境
        KettleEnvironment.init();

        // 初始化ktr
        String path = "E:\\yuanzheng-codebase\\m-framework\\kettle-git\\file\\test.ktr";
        TransMeta constantTrans = new TransMeta(path, new Variables());
        Trans trans = new Trans(constantTrans);
        trans.prepareExecution(new String[]{});
        trans.setPreview(true);
        trans.startThreads();
        trans.waitUntilFinished();

        // 关闭执行环境
        KettleEnvironment.shutdown();
        return null;
    }

    @Override
    public String jobSlave() throws Exception {

        // 初始化执行环境
        KettleEnvironment.init();

        // 节点服务参数
        SlaveServer remoteSlaveServer = new SlaveServer();
        remoteSlaveServer.setHostname("tcloud");
        remoteSlaveServer.setPort("8081");
        remoteSlaveServer.setUsername("cluster");
        remoteSlaveServer.setPassword("cluster");

        // 配置远程服务
        JobExecutionConfiguration jobExecutionConfiguration = new JobExecutionConfiguration();
        jobExecutionConfiguration.setRemoteServer(remoteSlaveServer);

        // 执行kjb
        String path = "E:\\yuanzheng-codebase\\m-framework\\kettle-git\\file\\test.kjb";
        FileSystemResource r = new FileSystemResource(path);
        JobMeta jobMeta = new JobMeta(r.getInputStream(), null, null);
        String carteObjectId = Job.sendToSlaveServer(jobMeta, jobExecutionConfiguration, null, null);

        // 获取状态信息
        System.out.println("carteObjectId=" + carteObjectId);
        SlaveServerJobStatus jobStatus;
        do {
            Thread.sleep(5000);
            jobStatus = remoteSlaveServer.getJobStatus(jobMeta.getName(), carteObjectId, 0);
            System.out.println(jobStatus.getResult());
        } while (jobStatus.isRunning());
        System.out.println(jobStatus.getResult());

        // 关闭执行环境
        KettleEnvironment.shutdown();
        return null;
    }

    @Override
    public String transSlave() throws Exception {

        // 初始化运行环境
        KettleEnvironment.init();

        // 节点服务参数
        SlaveServer remoteSlaveServer = new SlaveServer();
        remoteSlaveServer.setHostname("tcloud");
        remoteSlaveServer.setPort("8081");
        remoteSlaveServer.setUsername("cluster");
        remoteSlaveServer.setPassword("cluster");

        // 配置远程服务
        TransExecutionConfiguration transExecutionConfiguration = new TransExecutionConfiguration();
        transExecutionConfiguration.setRemoteServer(remoteSlaveServer);

        // 执行ktr
        String path = "E:\\yuanzheng-codebase\\m-framework\\kettle-git\\file\\test.ktr";
        TransMeta transMeta = new TransMeta(path, new Variables());
        String carteObjectId = Trans.sendToSlaveServer(transMeta, transExecutionConfiguration, null, null);

        // 获取状态信息
        System.out.println("carteObjectId=" + carteObjectId);
        SlaveServerTransStatus transStatus;
        do {
            Thread.sleep(5000);
            transStatus = remoteSlaveServer.getTransStatus(transMeta.getName(), carteObjectId, 0);
            System.out.println(transStatus.getResult());
        } while (transStatus.isRunning());
        System.out.println(transStatus.getResult());

        // 关闭执行环境
        KettleEnvironment.shutdown();
        return null;
    }

    private boolean checkOrderStatus(String status) {
        String[] tokens = {"Waiting", "Running", "Finished"};
        int offset = 0;
        for (String t : tokens) {
            while (status.startsWith(t, offset)) {
                offset += t.length();
            }
        }
        return offset == status.length();
    }
}
